package com.doitedu.core

import com.doitedu.utils.SparkUtil
import org.apache.spark.rdd.RDD

/**
 * @Date: 22.7.1 
 * @Author: HANGGE
 * @qq: 598196583
 * @Tips: 学大数据 ,到多易教育
 * @Description:
 */
object C17_AggregateByKey {
  def main(args: Array[String]): Unit = {


    val sc = SparkUtil.getSparkContext("reduceByKey")
 /*   val dataRDD = sc.textFile("data/word.txt")
    //coalesce 方法只能减少分布  不能增加分区 因为参数2 默认是false 不允许分区
    //  将参数2修改成true
    val wordsRDD = dataRDD.flatMap(_.split("\\s+"))

    val wordAndOne = wordsRDD.map((_, 1))
    val wordCnt: RDD[(String, Int)] = wordAndOne.reduceByKey(_ + _)
*/
 val dataRDD = sc.makeRDD(List(1, 2, 3, 4, 5, 6 ,7,8,9,10), 2)
    val kvRDD = dataRDD.map(i => {
      if (i % 2 == 0) {
        ("OU", i)
      } else {
        ("JI", i)
      }
    })


    /**
     * 两个参数类型列表
     * 参数类型列表1   默认值
     * 参数类型列表2
     *    1 分区内的计算逻辑
     *    2 分区间的计算逻辑
     *    hello tom
     *    hello tom
     *   [ hello 2
     *    tom  2 ]
     *   aggregateByKey: 按照key进行分组   对value进行计算
     *    1  2  3        OU  2    JI  1  3
     *    4  5  6        OU  4 6   JI  5
     *
     *   OU  2  | 4  6
     *   JI  1  3 | 5
     *   默认值参数分区内的局部运算
     *   (默认值)(分区内的计算逻辑, (分区间的计算逻辑))
     *
     *  统计各个组内的最大值数据
     */
    val max = kvRDD.aggregateByKey(0)((x1, x2) => if (x1 > x2) x1 else x2, (x1, x2) => if (x1 > x2) x1 else x2)
    max.foreach(println)

    /**
     * 用户信息
     *    按照城市分组
     *       获取每组的年龄最大的人的信息  , 如果最大年龄 <18 岁  18岁
     *
     */
    /*    val value: RDD[(String, String)] = kvRDD.aggregateByKey("DOIT")((x, y) => x + y, (str1, str2) => str1 + str2)
   /*
   (JI,DOIT13DOIT5)
   (OU,DOIT2DOIT46)
    */
    value.foreach(println)*/


  }

}
